package com.lightbend.akka.ddata

import akka.actor._
import akka.cluster._
import akka.persistence._
import scala.concurrent.duration._
import akka.cluster.sharding._

object Calculator {
  sealed trait Command
  case class Num(d: Double) extends Command
  case class Add(d: Double) extends Command
  case class Sub(d: Double) extends Command
  case class Mul(d: Double) extends Command
  case class Div(d: Double) extends Command
  case object ShowResult extends Command



  sealed trait Event
  case class SetResult(d: Any) extends Event

  def getResult(res: Double, cmd: Command) = cmd match {
    case Num(x) => x
    case Add(x) => res + x
    case Sub(x) => res - x
    case Mul(x) => res * x
    case Div(x) => {
      val _ = res.toInt / x.toInt //yield ArithmeticException when /0.00
      res / x
    }
    case _ => new ArithmeticException("Invalid Operation!")
  }

  case class State(result: Double) {

    def updateState(evt: Event): State = evt match {
      case SetResult(n) => copy(result = n.asInstanceOf[Double])
    }
  }

  case object Disconnect extends Command    //exit cluster

  def props = Props(new Calcultor)

}

class Calcultor extends PersistentActor with ActorLogging {
  import Calculator._
  val cluster = Cluster(context.system)

  var state: State = State(0)

  override def persistenceId: String = self.path.parent.name+"-"+self.path.name

  override def receiveRecover: Receive = {
    case evt: Event => state = state.updateState(evt)
    case SnapshotOffer(_,st: State) => state = state.copy(result =  st.result)
  }

  override def receiveCommand: Receive = {
    case Num(n) => persist(SetResult(getResult(state.result,Num(n))))(evt => state = state.updateState(evt))
    case Add(n) => persist(SetResult(getResult(state.result,Add(n))))(evt => state = state.updateState(evt))
    case Sub(n) => persist(SetResult(getResult(state.result,Sub(n))))(evt => state = state.updateState(evt))
    case Mul(n) => persist(SetResult(getResult(state.result,Mul(n))))(evt => state = state.updateState(evt))
    case Div(n) => persist(SetResult(getResult(state.result,Div(n))))(evt => state = state.updateState(evt))
    case ShowResult => log.info(s"Result on ${cluster.selfAddress.hostPort} is: ${state.result}")
    case Disconnect =>
      log.info(s"${cluster.selfAddress} is leaving cluster!!!")
      cluster.leave (cluster.selfAddress)

  }

  override def preRestart(reason: Throwable, message: Option[Any]): Unit = {
    log.info(s"Restarting calculator: ${reason.getMessage}")
    super.preRestart(reason, message)
  }
}

class CalcSupervisor extends Actor {
  def decider: PartialFunction[Throwable,SupervisorStrategy.Directive] = {
    case _: ArithmeticException => SupervisorStrategy.Resume
  }

  override def supervisorStrategy: SupervisorStrategy =
    OneForOneStrategy(maxNrOfRetries = 5, withinTimeRange = 5 seconds){
      decider.orElse(SupervisorStrategy.defaultDecider)
    }
  val calcActor = context.actorOf(Calculator.props,"calculator")

  override def receive: Receive = {
    case msg@ _ => calcActor.forward(msg)
  }

}

object CalculatorShard {
  import Calculator._

  case class CalcCommands(eid: String, msg: Command)  //user should use it to talk to shardregion
  val shardName = "calcShard"
  val getEntityId: ShardRegion.ExtractEntityId = {
    case CalcCommands(id,msg) => (id,msg)
  }
  val getShardId: ShardRegion.ExtractShardId = {
    case CalcCommands(id,_) => id.head.toString
  }
  def entityProps = Props(new CalcSupervisor)
}